#ifndef AVALON_SERVICEPROTOCOL_HPP
#define AVALON_SERVICEPROTOCOL_HPP
#include <vector>
#include <boost/call_traits.hpp>
#include <boost/preprocessor/repetition/enum_params.hpp>
#include <boost/preprocessor/repetition/enum_binary_params.hpp>
#include <boost/preprocessor/repetition/repeat.hpp>
#include <boost/preprocessor/iteration/local.hpp>
#include "Avalon/Base/Out.hpp"
#include "Avalon/IO/Buffer.hpp"
#include "Avalon/IO/Channel.hpp"
#include "Avalon/IO/Connection.hpp"
#include "Avalon/IO/ConnectionState.hpp"
#include "Avalon/IO/NotConnectedException.hpp"
#include "Avalon/IO/Writer.hpp"
#include "Avalon/Serialization/Deserializer.hpp"
#include "Avalon/Serialization/Serializer.hpp"
#include "Avalon/Services/Message.hpp"
#include "Avalon/Services/ServiceRequestException.hpp"
#include "Avalon/SignalHandling/TaskSignalHandler.hpp"
#include "Avalon/Threading/Async.hpp"

namespace Avalon {
namespace Services {

  /*! \class ServiceProtocol
      \brief Implements the service protocol for a Channel.
   */
  class ServiceProtocol {
    public:

      //! Constructs a ServiceProtocol.
      ServiceProtocol();

      virtual ~ServiceProtocol();

      //! Initializes the ServiceProtocol.
      /*!
        \param channel The Channel to implement the protocol for.
        \param serializer The Serializer to use for service Messages.
        \param deserializer The Deserializer to use for service Messages.
        \param slots The slots to call in response to a request.
        \param threadPool The ThreadPool to execute asynchronous tasks.
      */
      void Initialize(IO::Channel* channel,
        Serialization::Serializer* serializer,
        Serialization::Deserializer* deserializer,
        std::map<const std::type_info*, boost::function_base*>* slots,
        Threading::ThreadPool& threadPool);

      //! Returns the Channel the service protocol is being implemented for.
      IO::Channel* GetChannel();

      //! Starts the protocol.
      void Start();

      //! Stops the protocol.
      void Stop();

      //! Sends a Message.
      /*!
        \param message The Message to send.
      */
      void SendMessage(const Message& message);

      #define DECLARE_PARAMETER(z, n, q)                                       \
        BOOST_PP_COMMA_IF(n) typename boost::call_traits<                      \
          typename T::Parameter##n::Type>::param_type a##n

      #define LIST_PARAMETERS(z, n, q)                                         \
        BOOST_PP_COMMA_IF(n) a##n

      #define BOOST_PP_LOCAL_MACRO(n)                                          \
      template<typename T>                                                     \
      void SendRequest(BOOST_PP_REPEAT(n, DECLARE_PARAMETER,)                  \
          BOOST_PP_COMMA_IF(n)                                                 \
          typename Out<Threading::Async<typename T::ReturnType> > response) {  \
        typedef typename T::Request Request;                                   \
        typedef Threading::AsyncHolder<typename T::ReturnType> AsyncHolder;    \
        boost::unique_lock<boost::mutex> lock(m_mutex);                        \
        if(!m_isStarted || m_isStopping) {                                     \
          lock.unlock();                                                       \
          response->SetException(ServiceRequestException(                      \
            "Service protocol has stopped."));                                 \
          return;                                                              \
        }                                                                      \
        int requestId = m_nextRequestId;                                       \
        ++m_nextRequestId;                                                     \
        Request request(requestId BOOST_PP_COMMA_IF(n)                         \
          BOOST_PP_REPEAT(n, LIST_PARAMETERS,));                               \
        m_serializer->Shuttle("message", &request);                            \
        Threading::Async<void>* writeResult = new Threading::Async<void>();    \
        m_channel->GetWriter().Write(m_serializer->GetSerializedData(),        \
          Store(*writeResult));                                                \
        AsyncHolder* responseHolder = new AsyncHolder(response.Get());         \
        Threading::Async<void>::State writeState;                              \
        writeResult->SetFinishedSlot(m_signalHandler.GetSlot<                  \
          Threading::Async<void>::FinishedSlot>(m_writeSlot),                  \
          Store(writeState));                                                  \
        if(writeState == Threading::Async<void>::COMPLETE) {                   \
          delete writeResult;                                                  \
        } else if(writeState == Threading::Async<void>::EXCEPTION) {           \
          lock.unlock();                                                       \
          delete responseHolder;                                               \
          boost::exception_ptr exceptionCopy = writeResult->GetException();    \
          delete writeResult;                                                  \
          response->SetException(exceptionCopy);                               \
          return;                                                              \
        } else {                                                               \
          assert(writeState == Threading::Async<void>::PENDING);               \
          m_pendingWrites.push_back(writeResult);                              \
        }                                                                      \
        m_pendingRequests.insert(std::make_pair(requestId, responseHolder));   \
      }

      #define BOOST_PP_LOCAL_LIMITS (0, AVALON_SERVICE_PARAMETERS)
      #include BOOST_PP_LOCAL_ITERATE()
      #undef DECLARE_PARAMETER
      #undef LIST_PARAMETERS

    private:
      boost::mutex m_mutex;
      IO::Channel* m_channel;
      boost::scoped_ptr<Serialization::Serializer> m_serializer;
      boost::scoped_ptr<Serialization::Deserializer> m_deserializer;
      std::map<const std::type_info*, boost::function_base*>* m_slots;
      bool m_isStarted;
      bool m_isStopping;
      IO::Buffer m_readBuffer;
      Threading::Async<int> m_readResult;
      Threading::Async<int>::State m_readState;
      Threading::Async<int>::FinishedSlot m_writeSlot;
      std::deque<Threading::Async<void>*> m_pendingWrites;
      std::map<int, Threading::BaseAsyncHolder*> m_pendingRequests;
      int m_nextRequestId;
      boost::condition_variable m_isStoppedCondition;
      SignalHandling::TaskSignalHandler m_signalHandler;

      void CloseResources(boost::unique_lock<boost::mutex>& lock);
      void ProcessMessage(Message* message);
      void WriteToChannel(const IO::Buffer& buffer);
      void OnRead();
      void OnWrite();
  };

  /*! \class RequestToken
      \brief Represents a ServiceProtocol request.
   */
  template<typename Service>
  class RequestToken {
    public:

      //! Constructs a RequestToken.
      /*!
        \param protocol The ServiceProtocol making the request.
        \param requestId The request id.
      */
      RequestToken(ServiceProtocol* protocol, int requestId)
          : m_protocol(protocol),
            m_requestId(requestId) {}

      ~RequestToken() {}

      //! Returns the protocol that sent the request.
      ServiceProtocol* GetProtocol() const {
        return m_protocol;
      }

      //! Returns the Channel the request was received on.
      IO::Channel* GetChannel() const {
        return m_protocol->GetChannel();
      }

      //! Sends a response to this request.
      /*!
        \param response The response to send.
      */
      void SendResponse(const typename Service::ReturnType& response) const {
        m_protocol->SendMessage(typename Service::Response(m_requestId,
          response));
      }

    private:
      ServiceProtocol* m_protocol;
      int m_requestId;
  };
}
}

#endif // AVALON_SERVICEPROTOCOL_HPP
